package com.example.iotlink.mqtt;

import com.example.iotlink.bean.Receipt;
import com.example.iotlink.config.MqttProperties;
import com.example.iotlink.service.ITopicHandleStrategy;
import com.example.iotlink.service.ReceiptHandleStrategyFactory;
import com.example.iotlink.service.ReceiptStrategyContext;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.util.List;

/**
 * mqtt入站配置
 */
@Configuration
@Slf4j
public class MqttInboundConfiguration {
    @Autowired
    private MqttProperties mqttProperties;

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttPahoClientFactory) {
//        String[] inboundTopics = mqttProperties.getInbound().getTopics().split(",");
        List<String> topics = mqttProperties.getInbound().getTopics();
        String[] inboundTopics = topics.stream().toArray(String[]::new);

        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        mqttProperties.getInbound().getUrl(),
                        mqttProperties.getInbound().getClientId(),
                        mqttPahoClientFactory, inboundTopics);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
//                log.warn("mqtt_receivedTopic: {}", message.getHeaders().get("mqtt_receivedTopic"));

                ReceiptStrategyContext receiptStrategyContext = new ReceiptStrategyContext();
                ITopicHandleStrategy topicHandleStrategy = ReceiptHandleStrategyFactory.getReceiptHandleStrategy((String) message.getHeaders().get("mqtt_receivedTopic"));
                receiptStrategyContext.setTopicHandleStrategy(topicHandleStrategy);
                receiptStrategyContext.handleReceipt(Receipt.builder()
                        .message(message)
                        .topic((String) message.getHeaders().get("mqtt_receivedTopic"))
                        .build());


            }

        };
    }
}